//----------------------------------------------------------------
// include
//----------------------------------------------------------------
#include "thread_pool.h"





//----------------------------------------------------------------
// TaskQueue
//----------------------------------------------------------------
TaskQueue::TaskQueue()
{

}

//----------------------------------------------------------------
// ~TaskQueue
//----------------------------------------------------------------
TaskQueue::~TaskQueue()
{
    std::lock_guard<std::mutex> lk{mutex_};
    for (task_t *item = head_; item != nullptr;)
    {
        head_ = head_->next;
        if (item) delete item;
        item = head_;
    }
    cont_ = 0;
}

//----------------------------------------------------------------
// push_task
//----------------------------------------------------------------
void TaskQueue::push_task(task_func func, void *arg)
{
    task_t *task = new task_t;
    task->func = func;
    task->arg = arg;
    task->next = nullptr;
    std::lock_guard<std::mutex> lk{mutex_};
    if (head_ == tail_ && head_ == nullptr)
    {
        // 1. 空队列
        head_ = task;
        tail_ = task;
    }
    else
    {
        tail_->next = task;
        tail_ = task;
    }
    cont_++;
    std::cout << "push_task: task count " << cont_.load() << std::endl;
}

//----------------------------------------------------------------
// get_task
//----------------------------------------------------------------
TaskQueue::task_t* TaskQueue::get_task()
{
    if (!cont_.load()) return nullptr;
    std::lock_guard<std::mutex> lk{mutex_};

    task_t *task = head_;
    head_ = task->next;
    if (!head_) tail_ = nullptr;
    cont_--;
    std::cout << "get_task: task count " << cont_.load() << std::endl;
    return task;
}


//----------------------------------------------------------------
// ThreadPool
//----------------------------------------------------------------
ThreadPool::ThreadPool(int thread_cnt)
{
    task_queue_ = new TaskQueue();
    runing_.store(true);
    for (int i = 0; i < thread_cnt; i++)
    {
        std::thread *t = new std::thread(&ThreadPool::thread_task, this);
        thread_.push_back(t);
    }

    if (thread_cnt != thread_.size())
    {
        thread_destory();
    }
}

//----------------------------------------------------------------
// ~ThreadPool
//----------------------------------------------------------------
ThreadPool::~ThreadPool()
{
    if (task_queue_) delete task_queue_;
    thread_destory();
}

//----------------------------------------------------------------
// push_tash
//----------------------------------------------------------------
void ThreadPool::push_tash(task_func func, void *arg)
{
    if (thread_.size() == 0) return;
    task_queue_->push_task(func, arg);
    condv_.notify_one();
}

//----------------------------------------------------------------
// thread_task
//----------------------------------------------------------------
void ThreadPool::thread_task()
{
    while (runing_.load())
    {
        TaskQueue::task_t* task = task_queue_->get_task();
        if (task)
        {
            std::cout << "thread id[" << std::this_thread::get_id() << "]:" << "operate..." << std::endl;
            task->func(task->arg);
        }
        else
        {
            std::cout << "thread id[" << std::this_thread::get_id() << "]:" << "wait..." << std::endl;
            std::unique_lock<std::mutex> lk{mutex_};
            condv_.wait(lk);
        }
    }
}

//----------------------------------------------------------------
// thread_task
//----------------------------------------------------------------
void ThreadPool::thread_destory()
{
    if (thread_.size())
    {
        for (auto item = thread_.begin(); item != thread_.end(); item++)
        {
            condv_.notify_all();
            (*item)->join();
            delete (*item);
        }
    }
    thread_.clear();
}